package com.service.impl;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.lang.UUID;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.common.ErrorCode;
import com.constant.ChatType;
import com.domain.dto.UserSessionDto;
import com.domain.entity.ChatMessage;
import com.domain.entity.Team;
import com.domain.entity.User;
import com.domain.entity.UserSession;
import com.exception.BusinessException;
import com.mapper.UserSessionMapper;
import com.service.ChatMessageService;
import com.service.TeamService;
import com.service.UserSessionService;
import com.service.UsersService;
import com.ws.MessageManager;
import com.ws.SessionManager;
import com.ws.WebSocketSync;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.websocket.Session;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @author Administrator
 * @description 针对表【user_session】的数据库操作Service实现
 * @createDate 2024-05-24 16:35:46
 */
@Service
public class UserSessionServiceImpl extends ServiceImpl<UserSessionMapper, UserSession>
        implements UserSessionService {
    private static final long FIRST_SELECT_COUNT = 1;
    @Autowired
    private ChatMessageService chatMessageService;
    @Autowired
    private UsersService usersService;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Autowired
    private TeamService teamService;
    @Autowired
    private SessionManager sessionManager;
    @Autowired
    private MessageManager messageManager;


    @Override
    public UserSession createUserSession(Long sessionId, User loginUser, Integer chatType, Integer staicky, long deleteRecordTime, long lastSelectTime) {
        UserSession userSession = new UserSession();
        userSession.setSessionId(sessionId);
        userSession.setType(chatType);
        userSession.setIsSticky(staicky);
        userSession.setDeleteRecordTime(deleteRecordTime);
        userSession.setLastSelectTime(lastSelectTime);
        userSession.setUserId(loginUser.getId());
        return userSession;
    }

    /**
     * 获取所有的会话
     *
     * @param loginUser
     * @return
     */
    //这里可以开两个线程，返回最新消息，谁先获取到就返回谁，一个线程用于查redis和数据库，一个线程用于监听websocket
    @Override
    public List<UserSessionDto> getAllSession(User loginUser) {
        List<UserSession> userSessions = getAllUserSession(loginUser);
        List<UserSessionDto> userSessionDtos = userSessions.stream().map(this::getUserSessionDto).collect(Collectors.toList());
        //从websocket中同步数据
        userSessionDtos.forEach(this::syncWebSocketChatRecord);
        return userSessionDtos;
    }

    /**
     * 将usersession转换成Dto
     *
     * @param userSession
     * @return
     */
    private UserSessionDto getUserSessionDto(UserSession userSession) {
        UserSessionDto userSessionDto = userSessionToDto(userSession);

        //先从缓存中查找数据，再从数据库获取新消息，删除时间和上次查看时间的最大值。
        ZSetOperations<String, String> zSet = stringRedisTemplate.opsForZSet();
        String key;
        if (userSession.getType().equals(ChatType.PRIVATE)) {
            UUID uuid = messageManager.getPrivateChatKey(userSession.getSessionId(), userSession.getUserId());
            key = "private:" + uuid;
        } else {
            key = "team" + userSession.getSessionId();
        }
        long offsetTime = Math.max(userSession.getLastSelectTime(), userSession.getDeleteRecordTime());
        //获取新消息，和新消息数
        getNewMesAndNum(userSessionDto, zSet, key, offsetTime);
        long offset = System.currentTimeMillis();
        for (ChatMessage chatMessage : userSessionDto.getNewMessage()) {
            offset = Math.min(offset, chatMessage.getSendTime());
        }
        userSessionDto.setOffset(offset);
        // 获取会话名称和头像
        if (userSession.getType().equals(ChatType.PRIVATE)) {
            User user = usersService.getOne(new QueryWrapper<User>().eq("id", userSession.getSessionId()));
            userSessionDto.setSessionIcon(user.getGender());
            userSessionDto.setSessionName(user.getUsername());
        } else if (userSessionDto.getType().equals(ChatType.GROUP)) {
            Team team = teamService.getOne(new QueryWrapper<Team>().eq("id", userSession.getSessionId()));
            userSessionDto.setSessionIcon(team.getTeamIcon());
            userSessionDto.setSessionName(team.getName());
        }
        return userSessionDto;
    }

    private void syncWebSocketChatRecord(UserSessionDto userSessionDto) {
        Session session = sessionManager.getSession(userSessionDto.getUserId());
        if (WebSocketSync.hasSync(session)) {
            if (WebSocketSync.isUsed(session, userSessionDto.getSessionId())) {
                userSessionDto.setNewMessage(null);
            }
        }
    }

    private static UserSessionDto userSessionToDto(UserSession userSession) {
        UserSessionDto userSessionDto = new UserSessionDto();
        userSessionDto.setType(userSession.getType());
        userSessionDto.setIsSticky(userSession.getIsSticky());
        userSessionDto.setUserId(userSession.getUserId());
        userSessionDto.setSessionId(userSession.getSessionId());
        return userSessionDto;
    }

    private List<UserSession> getAllUserSession(User loginUser) {
        QueryWrapper<UserSession> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("user_id", loginUser.getId());
        List<UserSession> userSessionList = list(queryWrapper);
        if (CollectionUtil.isEmpty(userSessionList)) {
            throw new BusinessException(ErrorCode.PARAMS_ERROR, "没有会话!");
        }
        return userSessionList;
    }


    /**
     * 获取第一批新的消息和新消息数
     *
     * @param userSessionDto
     * @param zSet
     * @param key
     * @param offsetTime
     */
    private void getNewMesAndNum(UserSessionDto userSessionDto, ZSetOperations<String, String> zSet, String key, long offsetTime) {
        long minScore = getMinScore(zSet, key, offsetTime);
        List<ChatMessage> messagesfromCache = getChatMessagesfromCache(zSet, key, offsetTime);
        //缓存中有所有的新消息
        if (minScore < offsetTime) {
            //数据均来源于缓存
            userSessionDto.setNewMessage((ArrayList<ChatMessage>) messagesfromCache);
            userSessionDto.setNewMesCount(messagesfromCache.size());
        } else {//缓存中可能有部分新消息
            //缓存中的新消息记录数大于首次查询，新消息来源于缓存，新消息数来源于数据库
            if (messagesfromCache != null && messagesfromCache.size() >= FIRST_SELECT_COUNT) {
                userSessionDto.setNewMessage((ArrayList<ChatMessage>) messagesfromCache);
                getNewMesCountMySql(userSessionDto, messagesfromCache.size(), minScore, offsetTime);
                return;
            }
            SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH-mm-ss");
            String offsetFormat = dateFormat.format(new Date(offsetTime));
            QueryWrapper<ChatMessage> chatMessageQueryWrapper = new QueryWrapper<>();
            chatMessageQueryWrapper
                    .and(chatMessageQueryWrapper1 -> chatMessageQueryWrapper1
                            .eq("type", userSessionDto.getType())
                            .ge("send_time", offsetFormat))
                    .and(chatMessageQueryWrapper1 -> chatMessageQueryWrapper1
                            .eq("sender_id", userSessionDto.getUserId())
                            .eq("receive_id", userSessionDto.getSessionId())
                            .or(chatMessageQueryWrapper2 -> chatMessageQueryWrapper2
                                    .eq("sender_id", userSessionDto.getSessionId())
                                    .eq("receive_id", userSessionDto.getUserId())))
                    .orderBy(true, false, "send_time")
                    .last("limit " + FIRST_SELECT_COUNT);
            if (messagesfromCache != null && messagesfromCache.size() > 0) {
                chatMessageQueryWrapper.lt("send_time", dateFormat.format(new Date(minScore)));
            }
            getNewMesCountMySql(userSessionDto, messagesfromCache == null ? 0 : messagesfromCache.size(), minScore, offsetTime);
            List<ChatMessage> chatMessages = chatMessageService.list(chatMessageQueryWrapper);
            userSessionDto.setNewMessage((ArrayList<ChatMessage>) chatMessages);

        }
    }

    private void getNewMesCountMySql(UserSessionDto userSessionDto, Integer newChatRecordCount, long minScore, long offset) {
        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH-mm-ss");
        Date date = new Date(minScore);
        String minTimeFormat = dateFormat.format(date);
        Date date1 = new Date(offset);
        String offsetTimeFormat = dateFormat.format(date1);
        QueryWrapper<ChatMessage> chatMessageQueryWrapper = new QueryWrapper<>();
        QueryWrapper<ChatMessage> countWrapper = chatMessageQueryWrapper
                .and(chatMessageQueryWrapper1 -> chatMessageQueryWrapper1
                        .eq("sender_id", userSessionDto.getUserId())
                        .eq("receive_id", userSessionDto.getSessionId())
                        .or(chatMessageQueryWrapper2 -> chatMessageQueryWrapper2
                                .eq("sender_id", userSessionDto.getSessionId())
                                .eq("receive_id", userSessionDto.getUserId())))
                .eq("type", userSessionDto.getType())
                .ge("send_time", offsetTimeFormat)
                .le("send_time", minTimeFormat);
        int count = chatMessageService.count(countWrapper);
        userSessionDto.setNewMesCount((int) (count + newChatRecordCount));
    }

    /**
     * 从缓存中获取新消息
     *
     * @param zSet
     * @param key
     * @param offsetTime
     * @return
     */
    private static List<ChatMessage> getChatMessagesfromCache(ZSetOperations<String, String> zSet, String key, long offsetTime) {
        ArrayList<ChatMessage> chatMessageList = null;
        Set<String> chatRecord = zSet.reverseRangeByScore(key, offsetTime,
                System.currentTimeMillis(), 0, FIRST_SELECT_COUNT);
        if (!CollectionUtil.isEmpty(chatRecord)) {
            chatMessageList = (ArrayList<ChatMessage>) chatRecord.stream().map(s -> {
                return JSONUtil.toBean(s, ChatMessage.class);
            }).collect(Collectors.toList());
        }
        return chatMessageList;
    }

    private static long getMinScore(ZSetOperations<String, String> zSet, String key, long offsetTime) {
        Set<ZSetOperations.TypedTuple<String>> min = zSet.rangeByScoreWithScores(key, 0, 0);
        long minTime = offsetTime;
        if (!CollectionUtil.isEmpty(min)) {
            for (ZSetOperations.TypedTuple<String> stringTypedTuple : min) {
                long parseLong = Long.parseLong(Objects.requireNonNull(stringTypedTuple.getScore()).toString());
                if (parseLong != 0) {
                    minTime = parseLong;
                }
            }
        }
        return minTime;
    }
}




